// Copyright 2021-present The Atlas Authors. All rights reserved.
// This source code is licensed under the Apache 2.0 license found
// in the LICENSE file in the root directory of this source tree.

package dm

import (
	"context"
	"database/sql"
	"fmt"
	"math"
	"regexp"
	"strconv"
	"strings"

	"gitee.com/damengde/atlas/sql/internal/sqlx"
	"gitee.com/damengde/atlas/sql/schema"
)

// A diff provides a MySQL implementation for schema.Inspector.
type inspect struct{ *conn }

var _ schema.Inspector = (*inspect)(nil)

// InspectRealm returns schema descriptions of all resources in the given realm.
func (i *inspect) InspectRealm(ctx context.Context, opts *schema.InspectRealmOption) (*schema.Realm, error) {
	schemas, err := i.schemas(ctx, opts)
	if err != nil {
		return nil, err
	}
	if opts == nil {
		opts = &schema.InspectRealmOption{}
	}
	var (
		mode = sqlx.ModeInspectRealm(opts)
		r    = schema.NewRealm(schemas...).SetCharset(i.charset).SetCollation(i.collate)
	)
	if len(schemas) > 0 {
		if mode.Is(schema.InspectTables) {
			if err := i.inspectTables(ctx, r, nil); err != nil {
				return nil, err
			}
			sqlx.LinkSchemaTables(schemas)
		}
		if mode.Is(schema.InspectViews) {
			if err := i.inspectViews(ctx, r, nil); err != nil {
				return nil, err
			}
		}
		if mode.Is(schema.InspectFuncs) {
			if err := i.inspectFuncs(ctx, r, nil); err != nil {
				return nil, err
			}
		}
		if mode.Is(schema.InspectTriggers) {
			if err := i.inspectTriggers(ctx, r, nil); err != nil {
				return nil, err
			}
		}
	}
	return schema.ExcludeRealm(r, opts.Exclude)
}

// InspectSchema returns schema descriptions of the tables in the given schema.
// If the schema name is empty, the result will be the attached schema.
func (i *inspect) InspectSchema(ctx context.Context, name string, opts *schema.InspectOptions) (*schema.Schema, error) {
	schemas, err := i.schemas(ctx, &schema.InspectRealmOption{Schemas: []string{name}})
	if err != nil {
		return nil, err
	}
	switch n := len(schemas); {
	case n == 0:
		return nil, &schema.NotExistError{Err: fmt.Errorf("mysql: schema %q was not found", name)}
	case n > 1:
		return nil, fmt.Errorf("mysql: %d schemas were found for %q", n, name)
	}
	if opts == nil {
		opts = &schema.InspectOptions{}
	}
	var (
		mode = sqlx.ModeInspectSchema(opts)
		r    = schema.NewRealm(schemas...).SetCharset(i.charset).SetCollation(i.collate)
	)
	if mode.Is(schema.InspectTables) {
		if err := i.inspectTables(ctx, r, opts); err != nil {
			return nil, err
		}
		sqlx.LinkSchemaTables(schemas)
	}
	if mode.Is(schema.InspectViews) {
		if err := i.inspectViews(ctx, r, opts); err != nil {
			return nil, err
		}
	}
	if mode.Is(schema.InspectFuncs) {
		if err := i.inspectFuncs(ctx, r, opts); err != nil {
			return nil, err
		}
	}
	if mode.Is(schema.InspectTriggers) {
		if err := i.inspectTriggers(ctx, r, nil); err != nil {
			return nil, err
		}
	}
	return schema.ExcludeSchema(r.Schemas[0], opts.Exclude)
}

func (i *inspect) inspectTables(ctx context.Context, r *schema.Realm, opts *schema.InspectOptions) error {
	if err := i.tables(ctx, r, opts); err != nil {
		return err
	}
	for _, s := range r.Schemas {
		if len(s.Tables) == 0 {
			continue
		}
		if err := i.columns(ctx, s); err != nil {
			return err
		}
		if err := i.indexes(ctx, s); err != nil {
			return err
		}
	}
	return nil
}

// schemas returns the list of the schemas in the database.
func (i *inspect) schemas(ctx context.Context, opts *schema.InspectRealmOption) ([]*schema.Schema, error) {
	var (
		args  []any
		query = schemasQuery
	)
	if opts != nil {
		switch n := len(opts.Schemas); {
		case n == 1 && opts.Schemas[0] == "":
		case n == 1 && opts.Schemas[0] != "":
			query = fmt.Sprintf(schemasQueryArgs, "= ?")
			args = append(args, opts.Schemas[0])
		case n > 0:
			query = fmt.Sprintf(schemasQueryArgs, "IN ("+nArgs(len(opts.Schemas))+")")
			for _, s := range opts.Schemas {
				args = append(args, s)
			}
		}
	}
	rows, err := i.QueryContext(ctx, query, args...)
	if err != nil {
		return nil, fmt.Errorf("mysql: querying schemas: %w", err)
	}
	defer rows.Close()
	var schemas []*schema.Schema
	for rows.Next() {
		var name, charset string
		if err := rows.Scan(&name, &charset); err != nil {
			return nil, err
		}
		schemas = append(schemas, &schema.Schema{
			Name: name,
			Attrs: []schema.Attr{
				&schema.Charset{
					V: charset,
				},
			},
		})
	}
	return schemas, nil
}

func (i *inspect) tables(ctx context.Context, realm *schema.Realm, opts *schema.InspectOptions) error {
	var (
		args  []any
		query = fmt.Sprintf(tablesQuery, nArgs(len(realm.Schemas)))
	)
	for _, s := range realm.Schemas {
		args = append(args, s.Name)
	}
	if opts != nil && len(opts.Tables) > 0 {
		for _, t := range opts.Tables {
			args = append(args, t)
		}
		query = fmt.Sprintf(tablesQueryArgs, nArgs(len(realm.Schemas)), nArgs(len(opts.Tables)))
	}
	rows, err := i.QueryContext(ctx, query, args...)
	if err != nil {
		return err
	}
	defer rows.Close()
	for rows.Next() {
		var (
			tSchema, name, charset sql.NullString
		)
		if err := rows.Scan(&tSchema, &name, &charset); err != nil {
			return fmt.Errorf("scan table information: %w", err)
		}
		if !sqlx.ValidString(tSchema) || !sqlx.ValidString(name) {
			return fmt.Errorf("invalid schema or table name: %q.%q", tSchema.String, name.String)
		}
		s, ok := realm.Schema(tSchema.String)
		if !ok {
			return fmt.Errorf("schema %q was not found in realm", tSchema.String)
		}
		t := &schema.Table{Name: name.String}
		s.AddTables(t)
		if sqlx.ValidString(charset) {
			t.Attrs = append(t.Attrs, &schema.Charset{
				V: charset.String,
			})
		}
	}
	return rows.Close()
}

// columns queries and appends the columns of the given table.
func (i *inspect) columns(ctx context.Context, s *schema.Schema) error {
	query := columnsQuery
	rows, err := i.querySchema(ctx, query, s)
	if err != nil {
		return fmt.Errorf("mysql: query schema %q columns: %w", s.Name, err)
	}
	defer rows.Close()
	for rows.Next() {
		if err := i.addColumn(s, rows); err != nil {
			return fmt.Errorf("mysql: %w", err)
		}
	}
	return rows.Err()
}

// addColumn scans the current row and adds a new column from it to the table.
func (i *inspect) addColumn(s *schema.Schema, rows *sql.Rows) error {
	var table, name, typ, nullable, key, defaults sql.NullString
	var numericPrecision, lgth, numericScale int
	if err := rows.Scan(&table, &name, &typ, &lgth, &nullable, &key, &defaults, &numericPrecision, &numericScale); err != nil {
		return err
	}
	t, ok := s.Table(table.String)
	if !ok {
		return fmt.Errorf("table %q was not found in schema", table.String)
	}
	c := &schema.Column{
		Name: name.String,
		Type: &schema.ColumnType{
			Raw:  typ.String,
			Null: nullable.String == "Y",
		},
	}
	ct, err := ParseType(c.Type.Raw, lgth, numericPrecision, numericScale)
	if err != nil {
		return fmt.Errorf("parse %q.%q type %q: %w", t.Name, c.Name, c.Type.Raw, err)
	}
	c.Type.Type = ct
	if defaults.Valid {
		c.Default = i.myDefaultExpr(c, defaults.String, nil)
	}
	t.AddColumns(c)
	return nil
}

// indexes queries and appends the indexes of the given table.
func (i *inspect) indexes(ctx context.Context, s *schema.Schema) error {
	query := i.indexQuery()
	rows, err := i.querySchema(ctx, query, s)
	if err != nil {
		return fmt.Errorf("mysql: query schema %q indexes: %w", s.Name, err)
	}
	defer rows.Close()
	if err := i.addIndexes(s, rows); err != nil {
		return err
	}
	return rows.Err()
}

// addIndexes scans the rows and adds the indexes to the table.
func (i *inspect) addIndexes(s *schema.Schema, rows *sql.Rows) error {
	hasPK := make(map[*schema.Table]bool)
	for rows.Next() {
		var (
			table, name, indexType, column string
		)
		if err := rows.Scan(&table, &name, &indexType, &column); err != nil {
			return fmt.Errorf("mysql: scanning indexes for schema %q: %w", s.Name, err)
		}
		t, ok := s.Table(table)
		if !ok {
			return fmt.Errorf("table %q was not found in schema", table)
		}
		// Ignore primary keys.
		if indexType == "P" {
			hasPK[t] = true
			var cs []*schema.Column
			cns := strings.Split(column, ",")
			for i := range cns {
				for _, c := range t.Columns {
					if cns[i] == c.Name {
						cs = append(cs, c)
					}
				}
			}
			t.PrimaryKey = schema.NewPrimaryKey(cs...)
			continue
		}
		idx, ok := t.Index(name)
		if !ok {
			idx = schema.NewIndex(name).
				SetUnique(indexType == "U").
				AddAttrs(&IndexType{T: indexType})
			t.AddIndexes(idx)
		}
	}
	for _, t := range s.Tables {
		if !hasPK[t] && t.PrimaryKey != nil {
			t.PrimaryKey = nil
		}
	}
	return nil
}

// fks queries and appends the foreign keys of the given table.
func (i *inspect) fks(ctx context.Context, s *schema.Schema) error {
	return nil
}

// checks queries and appends the check constraints of the given table.

// indexQuery returns the query to retrieve the indexes of the given table.
func (i *inspect) indexQuery() string {
	query := indexesNoCommentQuery
	return query
}

// extraAttr is a parsed version of the information_schema EXTRA column.
type extraAttr struct {
	autoinc          bool
	onUpdate         string
	generatedType    string
	defaultGenerated bool
}

var (
	reGenerateType = regexp.MustCompile(`(?i)^(stored|persistent|virtual) generated$`)
	reTimeOnUpdate = regexp.MustCompile(`(?i)^(?:default_generated )?on update (current_timestamp(?:\(\d?\))?)$`)
)

// parseExtra returns a parsed version of the EXTRA column
// from the INFORMATION_SCHEMA.COLUMNS table.
func parseExtra(extra string) (*extraAttr, error) {
	attr := &extraAttr{}
	switch el := strings.ToLower(extra); {
	case el == "", el == "null":
	case el == defaultGen:
		attr.defaultGenerated = true
		// The column has an expression default value,
		// and it is handled in Driver.addColumn.
	case el == autoIncrement:
		attr.autoinc = true
	case reTimeOnUpdate.MatchString(extra):
		attr.onUpdate = reTimeOnUpdate.FindStringSubmatch(extra)[1]
	case reGenerateType.MatchString(extra):
		attr.generatedType = reGenerateType.FindStringSubmatch(extra)[1]
	default:
		return nil, fmt.Errorf("unknown extra column attribute %q", extra)
	}
	return attr, nil
}

// showCreate sets and fixes schema elements that require information from
// the 'SHOW CREATE' command.
func (i *inspect) showCreate(ctx context.Context, s *schema.Schema) error {
	for _, t := range s.Tables {
		st, ok := popShow(t)
		if !ok {
			continue
		}
		c, err := i.createStmt(ctx, t)
		if err != nil {
			return err
		}
		st.setIndexParser(c)
		if err := st.setAutoInc(t, c); err != nil {
			return err
		}
	}
	return nil
}

var reAutoinc = regexp.MustCompile(`(?i)\s*AUTO_INCREMENT\s*=\s*(\d+)\s*`)

// createStmt loads the CREATE TABLE statement for the table.
func (i *inspect) createStmt(ctx context.Context, t *schema.Table) (*CreateStmt, error) {
	c := &CreateStmt{}
	b := &sqlx.Builder{QuoteOpening: '`', QuoteClosing: '`'}
	rows, err := i.QueryContext(ctx, b.P("SHOW CREATE TABLE").Table(t).String())
	if err != nil {
		return nil, fmt.Errorf("query CREATE TABLE %q: %w", t.Name, err)
	}
	if err := sqlx.ScanOne(rows, &sql.NullString{}, &c.S); err != nil {
		return nil, fmt.Errorf("scan CREATE TABLE %q: %w", t.Name, err)
	}
	t.Attrs = append(t.Attrs, c)
	return c, nil
}

var reCurrTimestamp = regexp.MustCompile(`(?i)^current_timestamp(?:\(\d?\))?$`)

// myDefaultExpr returns the correct schema.Expr based on the column attributes for MySQL.
func (i *inspect) myDefaultExpr(c *schema.Column, x string, attr *extraAttr) schema.Expr {
	// In MySQL, the DEFAULT_GENERATED indicates the column has an expression default value.

	switch c.Type.Type.(type) {
	case *schema.BinaryType:
		// MySQL v8 uses Hexadecimal representation.
		if isHex(x) {
			return &schema.Literal{V: x}
		}
	case *BitType, *schema.BoolType, *schema.IntegerType, *schema.DecimalType, *schema.FloatType:
		return &schema.Literal{V: x}
	case *schema.TimeType:
		// "current_timestamp" is exceptional in old versions
		// of MySQL for timestamp and datetime data types.
		if reCurrTimestamp.MatchString(x) {
			return &schema.RawExpr{X: x}
		}
	}
	return &schema.Literal{V: quote(x)}
}

// parseColumn returns column parts, size and signed-info from a MySQL type.
func parseColumn(typ string) (parts []string, size int, unsigned bool, err error) {
	return []string{typ}, math.MaxUint32, unsigned, nil
}

// hasNumericDefault reports if the given type has a numeric default value.
func hasNumericDefault(t schema.Type) bool {
	switch t.(type) {
	case *BitType, *schema.BoolType, *schema.IntegerType, *schema.DecimalType, *schema.FloatType:
		return true
	}
	return false
}

func isHex(x string) bool { return len(x) > 2 && strings.ToLower(x[:2]) == "0x" }

// marDefaultExpr returns the correct schema.Expr based on the column attributes for MariaDB.
func (i *inspect) marDefaultExpr(c *schema.Column, x string) schema.Expr {
	// Unlike MySQL, NULL means default to NULL or no default.
	if x == "NULL" {
		return nil
	}
	// In this case, we need to manually check if the expression is literal, or fallback to raw expression.
	switch c.Type.Type.(type) {
	case *BitType:
		// Bit literal values. See https://mariadb.com/kb/en/binary-literals.
		if strings.HasPrefix(x, "b'") && strings.HasSuffix(x, "'") {
			return &schema.Literal{V: x}
		}
	case *schema.BoolType, *schema.IntegerType, *schema.DecimalType, *schema.FloatType:
		if _, err := strconv.ParseFloat(x, 64); err == nil {
			return &schema.Literal{V: x}
		}
	case *schema.TimeType:
		// "current_timestamp" is exceptional in old versions
		// of MySQL (i.e. MariaDB in this case).
		if strings.ToLower(x) == currentTS {
			return &schema.RawExpr{X: x}
		}
	}

	return &schema.RawExpr{X: sqlx.MayWrap(x)}
}

func (i *inspect) querySchema(ctx context.Context, query string, s *schema.Schema) (*sql.Rows, error) {
	// Number of times the schema name is parameterized.
	args := make([]any, strings.Count(query, "?"))
	for i := range args {
		args[i] = s.Name
	}
	for _, t := range s.Tables {
		args = append(args, t.Name)
	}
	return i.QueryContext(ctx, fmt.Sprintf(query, nArgs(len(s.Tables))), args...)
}

func nArgs(n int) string { return strings.Repeat("?, ", n-1) + "?" }

const (

	// Query current session schemas.
	schemasQuery = `SELECT SF_GET_SCHEMA_NAME_BY_ID(CURRENT_SCHID),(select UNICODE ()) AS ct`

	// Query to list specific database schemas.
	schemasQueryArgs = "SELECT DISTINCT OBJECT_NAME AS sn,(select UNICODE ()) AS ct FROM ALL_OBJECTS WHERE OBJECT_TYPE = 'SCH' AND OBJECT_NAME %s"

	// Query to list table columns.
	columnsQuery = "SELECT TABLE_NAME,COLUMN_NAME,DATA_TYPE,DATA_LENGTH,NULLABLE,COLUMN_ID,DATA_DEFAULT,isnull(DATA_PRECISION,0),DATA_SCALE FROM DBA_TAB_COLUMNS  WHERE OWNER = ? AND TABLE_NAME IN (%s) ORDER BY TABLE_NAME"

	indexesNoCommentQuery = `
		SELECT a.TABLE_NAME,a.CONSTRAINT_NAME,a.CONSTRAINT_TYPE,b.COLUMN_NAME
  		FROM DBA_CONSTRAINTS a 
  		JOIN ALL_CONS_COLUMNS b 
    		ON a.CONSTRAINT_NAME=b.CONSTRAINT_NAME 
 		Where a.OWNER= ? AND a.TABLE_NAME IN (%s)
		`
	tablesQuery     = `select OWNER,TABLE_NAME,(select UNICODE ()) AS ct  from all_tables where OWNER IN (%s)`
	tablesQueryArgs = `select OWNER,TABLE_NAME,(select UNICODE ()) AS ct  from all_tables where OWNER IN (%s) AND TABLE_NAME IN (%s)`
)

type (
	// AutoIncrement attribute for columns with "AUTO_INCREMENT" as a default.
	// V represent an optional start value for the counter.
	AutoIncrement struct {
		schema.Attr
		V int64
	}

	// CreateOptions attribute for describing extra options used with CREATE TABLE.
	CreateOptions struct {
		schema.Attr
		V string
	}

	// CreateStmt describes the SQL statement used to create a table.
	CreateStmt struct {
		schema.Attr
		S string
	}

	// Engine attribute describes the storage engine used to create a table.
	Engine struct {
		schema.Attr
		V       string // InnoDB, MyISAM, etc.
		Default bool   // The default engine used by the server.
	}

	// OnUpdate attribute for columns with "ON UPDATE CURRENT_TIMESTAMP" as a default.
	OnUpdate struct {
		schema.Attr
		A string
	}

	// SubPart attribute defines an option index prefix length for columns.
	SubPart struct {
		schema.Attr
		Len int
	}

	// Enforced attribute defines the ENFORCED flag for CHECK constraint.
	Enforced struct {
		schema.Attr
		V bool // V indicates if the CHECK is enforced or not.
	}

	// The DisplayWidth represents a display width of an integer type.
	DisplayWidth struct {
		schema.Attr
		N int
	}

	// The ZeroFill represents the ZEROFILL attribute which is
	// deprecated for MySQL version >= 8.0.17.
	ZeroFill struct {
		schema.Attr
		A string
	}

	// IndexType represents an index type.
	IndexType struct {
		schema.Attr
		T string // BTREE, HASH, FULLTEXT, SPATIAL, RTREE
	}

	// IndexParser defines the parser plugin used
	// by a FULLTEXT index.
	IndexParser struct {
		schema.Attr
		P string // Name of the parser plugin. e.g., ngram or mecab.
	}

	// BitType represents the type bit.
	BitType struct {
		schema.Type
		T    string
		Size int
	}

	// SetType represents a set type.
	SetType struct {
		schema.Type
		Values []string
	}

	// putShow is an intermediate table attribute used
	// on inspection to indicate if the 'SHOW TABLE' is
	// required and for what.
	showTable struct {
		schema.Attr
		// AUTO_INCREMENT value due to missing value in information_schema.
		auto *AutoIncrement
		// FULLTEXT indexes that might have custom parser.
		idxs []*schema.Index
	}
)

// addIndex adds an index to the list of indexes
// that needs further processing.
func (s *showTable) addFullText(idx *schema.Index) {
	s.idxs = append(s.idxs, idx)
}

// setAutoInc extracts the updated AUTO_INCREMENT from CREATE TABLE.
func (s *showTable) setAutoInc(t *schema.Table, c *CreateStmt) error {
	if s.auto == nil {
		return nil
	}
	if sqlx.Has(t.Attrs, &AutoIncrement{}) {
		return fmt.Errorf("unexpected AUTO_INCREMENT attributes for table: %q", t.Name)
	}
	matches := reAutoinc.FindStringSubmatch(c.S)
	if len(matches) != 2 {
		return nil
	}
	v, err := strconv.ParseInt(matches[1], 10, 64)
	if err != nil {
		return err
	}
	s.auto.V = v
	t.Attrs = append(t.Attrs, s.auto)
	return nil
}

// reIndexParser matches the parser name from the index definition.
var reIndexParser = regexp.MustCompile("/\\*!50100 WITH PARSER `([^`]+)` \\*/")

// setIndexParser updates the FULLTEXT parser from CREATE TABLE statement.
func (s *showTable) setIndexParser(c *CreateStmt) {
	b := (&sqlx.Builder{QuoteOpening: '`', QuoteClosing: '`'}).P("FULLTEXT KEY")
	for _, idx := range s.idxs {
		bi := b.Clone().Ident(idx.Name).Wrap(func(b *sqlx.Builder) {
			b.MapComma(idx.Parts, func(i int, b *sqlx.Builder) {
				// We expect column names only, as functional
				// fulltext indexes are not supported by MySQL.
				if idx.Parts[i].C != nil {
					b.Ident(idx.Parts[i].C.Name)
				}
			})
		})
		i := strings.Index(c.S, bi.String())
		if i == -1 || i+bi.Len() >= len(c.S) {
			continue
		}
		i += bi.Len()
		j := strings.Index(c.S[i:], "\n")
		if j == -1 {
			continue
		}
		// The rest of the line holds index, algorithm and lock options.
		if matches := reIndexParser.FindStringSubmatch(c.S[i : i+j]); len(matches) == 2 {
			idx.AddAttrs(&IndexParser{P: matches[1]})
		}
	}
}

func putShow(t *schema.Table) *showTable {
	for i := range t.Attrs {
		if s, ok := t.Attrs[i].(*showTable); ok {
			return s
		}
	}
	s := &showTable{}
	t.Attrs = append(t.Attrs, s)
	return s
}

func popShow(t *schema.Table) (*showTable, bool) {
	for i := range t.Attrs {
		if s, ok := t.Attrs[i].(*showTable); ok {
			t.Attrs = append(t.Attrs[:i], t.Attrs[i+1:]...)
			return s, true
		}
	}
	return nil, false
}
